package run;

import com.SpringBootApps;
import io.rsocket.metadata.WellKnownMimeType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import utils.Utils;

import java.net.URI;
import java.time.Duration;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest(classes = {StartUpForRSocketClientB.Dummy.class},
        webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)
public class StartUpForRSocketClientB {
    Object lock = new Object();
    @Autowired
    GenericApplicationContext context;

    private  RSocketRequester requester;
    private  RSocketRequester requesterWithWebSocket;


    // request ,
    @BeforeEach
    @SuppressWarnings("ConstantConditions")
    public  void setupOnce() {
        MimeType metadataMimeType = MimeTypeUtils.parseMimeType(
                WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString());

        requester = RSocketRequester.builder()
                .metadataMimeType(metadataMimeType)
                .rsocketStrategies(context.getBean(RSocketStrategies.class))
                .tcp("localhost", 7000);

        requesterWithWebSocket = RSocketRequester.builder()
                .metadataMimeType(metadataMimeType)
                .rsocketStrategies(context.getBean(RSocketStrategies.class))
                .websocket(URI.create("ws://localhost:7000/ws"));

    }
    private static final MimeType FOO_MIME_TYPE = MimeTypeUtils.parseMimeType("messaging/x.foo");
    @Test
    public void metadataPushWithWebSocket() {
        Flux.just("bar", "baz")
                .concatMap(s -> requesterWithWebSocket.route("foo-updates").metadata(s, FOO_MIME_TYPE).sendMetadata())
                .blockLast();
    }

    @Test
    public void metadataPush() {
        Flux.just("bar", "baz")
                .concatMap(s -> requester.route("foo-updates").metadata(s, FOO_MIME_TYPE).sendMetadata())
                .blockLast();
    }
    @Test
    public void echoWithWebSocket(){
        // send the request .
        Flux<String> result = Flux.range(1, 3).concatMap(i ->
                requesterWithWebSocket.route("echo").data("Hello " + i).retrieveMono(String.class));

        StepVerifier.create(result)
                .expectNext("Hello 1").expectNext("Hello 2").expectNext("Hello 3")
                .expectComplete()
                .verify(Duration.ofSeconds(5));

    }
    @Test
    public void echo(){
        // send the request .
        Flux<String> result = Flux.range(1, 3).concatMap(i ->
                requester.route("echo").data("Hello " + i).retrieveMono(String.class));

        StepVerifier.create(result)
                .expectNext("Hello 1").expectNext("Hello 2").expectNext("Hello 3")
                .expectComplete()
                .verify(Duration.ofSeconds(5));

    }
    @Test
    public void verify() throws InterruptedException {


        Utils.print(context);
        synchronized (lock) {
            lock.wait();
        }
    }

    @Import(value = {SpringBootApps.class})
    @Configuration
    public static class Dummy {

        @Bean
        public RSocketStrategies rsocketStrategies() {
            return RSocketStrategies.builder()
                    .metadataExtractorRegistry(registry ->
                            registry.metadataToExtract(FOO_MIME_TYPE, String.class, "foo"))
                    .build();
        }

    }

}
